package com.example.channel

import com.google.gson.Gson
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.selects.select
import org.junit.Test
import java.io.File


/**
 * 多路复用技术
 * 数据通信系统或者计算机网络中，
 * 传输媒体的贷款或者容量往往大于传输单一信号的需求。为了有效的利用通信线路，
 * 希望一个信道同时传输多路信号，这就是所谓的多路复用技术（Multiplexing）
 *
 * @author zhouronghua
 * @time 2022/1/8 9:06 下午
 */
class ChannelMultiPlexTest {

    /**
     * 数据类：用户对象
     */
    data class User(val name: String, val ID: String)

    data class Response<T>(val value: T, val isLocal: Boolean)

    suspend fun CoroutineScope.getUserFromLocal(name: String) = async {
        delay(1000)
        File("/Users/qwe/workspace/projects/cdel/kotlin_demo/userinfo.txt").readText().let {
            Gson().fromJson(it, User::class.java)
        }
    }

    suspend fun CoroutineScope.getUserFromNet(name: String) = async {
        delay(500)
        User(name, "1000000")
    }

    /**
     * 采用select进行多路复用
     * 对异步协程进行多路复用
     *
     * @author zhouronghua
     * @time 2022/1/8 9:07 下午
     */
    @Test
    fun test_multi_plex_select_await() = runBlocking {
        GlobalScope.launch {
            val localRequest = getUserFromLocal(" 李磊")
            val netRequest = getUserFromNet(" 李磊")
            // 两个异步协程任务进行多路复用，接收到最早完成的协程任务信息
            val response = select<Response<User>> {
                val selectClause1 = localRequest.onAwait {
                    // 协程返回的User对象，封装为Response返回
                    Response(it, true)
                }
                val selectClause2 = netRequest.onAwait {
                    Response(it, false)
                }
            }
            response?.let {
                println(it)
            }
        }.join()
    }

    /**
     * 两个channel发送消息，channel1每100ms发送一个；
     * channel2每150毫秒发送一个，接收端采用多路复用进行接收
     *
     * @author zhouronghua
     * @time 2022/1/8 9:54 下午
     */
    @Test
    fun test_multi_plx_channel() = runBlocking {
        val channels = listOf(Channel<String>(), Channel<String>())
        val job1 = GlobalScope.launch(Dispatchers.IO) {
            for (i in 1..30) {
                delay(100)
                println("channel 1 send $i")
                channels[0].send("channel 1 send $i")
            }
            channels[0].close()
        }
        val job2 = GlobalScope.launch(Dispatchers.IO) {
            for (i in 1..30) {
                delay(150)
                println("channel 2 send $i")
                channels[1].send("channel 2 == send $i")
            }
            channels[1].close()
        }
        while (true) {
            try {
                val result = select<String?> {
                    channels.onEach {
                        it.onReceive<String?> {
                            println("receive element:  $it")
                            it
                        }
                    }
                }
                delay(100)
                println(result)
            } catch (e: Throwable) {
                println("Caught: $e")
                break
            } finally {
                if (channels[0].isClosedForSend && channels[1].isClosedForSend) {
                    break
                }
            }
        }
        joinAll(job1, job2)
        delay(500)
    }

    /**
     * 测试SelectClause0
     * 没有返回值也没有参数
     *
     * @author zhouronghua
     * @time 2022/1/9 7:35 下午
     */
    @Test
    fun test_SelectClause0() = runBlocking {
        println("测试无返回值的多路复用onJoin")
        val job1 = GlobalScope.launch(Dispatchers.IO) {
            delay(100)
            println("Job 1 finish")
        }
        val job2 = GlobalScope.launch(Dispatchers.IO) {
            delay(120)
            println("Job 2 finish")
        }
        select<Unit> {
            job1.onJoin { println("Job1 onJoin") }
            job2.onJoin { println("Job2 onJoin") }
        }
    }

    /**
     * 测试SelectClause2
     * Channel.onSend发送内容，以及发送的结果回调
     * 说明：在一个协程中，两个channel进行多路发送，
     * 在另外两个协程中进行接收
     *
     * @author zhouronghua
     * @time 2022/1/9 7:44 下午
     */
    @Test
    fun test_selectClause2() = runBlocking {
        // 创建两个Channel
        val channels = listOf(Channel<Int>(), Channel<Int>())
        println(channels)
        // 创建一个协程对两个通道进行多路发送
        val job1 = launch(Dispatchers.IO) {
            println("协程中select发送")
            select<Unit?> {
                launch {
                    channels[0].onSend(200) {
                        println("send on channel0 $it")
                    }
                }
                launch {
                    channels[1].onSend(100) {
                        println("send on channel1 $it")
                    }
                }
            }
        }
        // 通过两个协程接收
        val job2 = GlobalScope.launch {
            println("channel 0 接收 ${channels[0].receive()}")
        }
        // 通过两个协程接收
        val job3 = GlobalScope.launch {
            println("channel 1 接收  ${channels[1].receive()}")
        }
        joinAll(job1, job2, job3)
        delay(1000)
    }

//    fun test_flow_multi_plex() = runBlocking {
//
//    }
}